package com.jida.hadoop.mr.PV;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 
public class DxPathStatisticDriver {
    public static void main(String[] args){
        try {
            //创建配置信息
            Configuration conf = new Configuration();
            //map内存设置(默认1024)
            conf.set("mapreduce.map.memory.mb", "5120");
            //检查超时(默认600000ms)，由于集群环境不稳定，有超时现象，所以设置为不检查，但需要保证程序无死循环
            conf.set("mapreduce.task.timeout", "0");
            //集群机器少的时候可以设置：客户端在写失败的时候，不使用更换策略
            conf.set("dfs.client.block.write.replace-datanode-on-failure.enable","true");
            conf.set("dfs.client.block.write.replace-datanode-on-failure.policy","NEVER");
            // 创建任务
            Job job = Job.getInstance(conf, "Dx_File_pv_staticstic");
            // 打成jar包运行，这句话是关键
            job.setJarByClass(DxPathStatisticDriver.class);
 
            // 设置最小输入分片大小(默认0)128M
            TextInputFormat.setMinInputSplitSize(job,1024*1024*128L); // 设置最小输入分片大小(默认0)128M
 
            // 1.2 设置自定义Mapper类、设置map函数输出数据的key和value的类型
            job.setMapperClass(MapPvNumMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
 
            job.setReducerClass(PvNumReduce.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(NullWritable.class);
 
//            job.setNumReduceTasks(3);
            //输入路径，也就是01数据清洗后的输出数据
            FileInputFormat.addInputPath(job, new Path("file:///D:\\测试数据\\dx\\02.DxFileMatch/part-m-00*"));
            //输出路径
            FileOutputFormat.setOutputPath(job, new Path("file:///D:\\测试数据\\dx\\03.DxPathStatistic"));
            //提交作业 判断退出条件（0正常退出，1非正常退出）
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
